<?php
namespace App\Common\Lib\Com\Kafka;

use App\Common as com;
use App\Common\Lib\Com\Kafka\Producer as thisProducer;
use longlang\phpkafka\Producer\Producer as kafkaProducer;
use longlang\phpkafka\Producer\ProducerConfig;

class Producer extends com\Lib\LibBase {

	private ProducerConfig $productConfig;

	private kafkaProducer $producer;

	public function __construct( ProducerConfig $productConfig ) {
		$this->productConfig = $productConfig;
		//
		$this->producer = new kafkaProducer( $this->productConfig );
		//
		parent::__construct();
	}

	public function send( string $topic, ?string $value, ?string $key = null, array $headers = [], ?int $partitionIndex = null, ?int $brokerId = null ) {
		$this->producer->send( $topic, $value, $key, $headers, $partitionIndex, $brokerId );
	}

	// ===============================
	//
	//
	// 下面为静态方法
	//
	//
	// ===============================

	/**
	 * @return static
	 */
	public static function newOfDefault():self {
		$config = new ProducerConfig();
		$config->setBootstrapServer( '192.168.147.128:9092' );
		$config->setUpdateBrokers( true );
		$config->setAcks( -1 );
		// $config->setAutoCreateTopic( true );
		// auto.create.topics.enable
		//
		return self::make( $config );
	}

}
